package com.its.common.disruptor.use;


import com.its.common.disruptor.consumer.BaseMessageConsumer;
import com.its.common.disruptor.dto.TranslatorData;
import com.its.common.disruptor.factory.RingBufferWorkerPoolFactory;
//import io.reactivex.rxjava3.schedulers.Schedulers;

import java.util.UUID;

/**
 * @Author xiaxp
 * @Date 2021-11-25 23:13
 * @Description
 */
public class Test {
    public static void main(String[] args) throws InterruptedException {
        int consumerCnt = 16;
        BaseMessageConsumer[] consumers = new BaseMessageConsumer[consumerCnt];

        for (int i = 0; i < consumers.length; i++){
            BaseMessageConsumer consumer = new MessageConsumerImpl("code:serverId:"+i);
            consumers[i] = consumer;
        }

//        for (int i = 0; i < consumers.length; i++){
//            BaseMessageConsumer consumer = new TimedAndMaxCountOnceMessageConsumerImpl("code:serverId:"+i,
//                    1000,1000*2L);
//            consumers[i] = consumer;
//        }

        try {
            RingBufferWorkerPoolFactory.getInstance().initAndStart(consumers);
        } catch (Exception e) {
            e.printStackTrace();
        }

        firstProduce();
        secondProduce(2);
    }

    private static void firstProduce() throws InterruptedException {
        System.out.println("--------------------- 开始生产数据 -----------------------");
        long start = System.currentTimeMillis();
        for (int i = 0; i < 100000; i++) {
            if(i % 100 == 0){
                Thread.sleep(28L);
            }
            TranslatorData request = new TranslatorData();


            String id = UUID.randomUUID() + " --1- "+ i;
            request.setId(id);

            TestObject object = new TestObject(id,i);

            request.setId(id);
            request.setMessage(new TestObject[]{object});

            // 作为生产，加入 RingBuffer 处理
            RingBufferWorkerPoolFactory.getInstance().produceData(request);
        }
        long end = System.currentTimeMillis();
        System.out.println("--------------------- 结束生产数据 -----------------------" + (end -start));

    }

    private static void secondProduce(int delayMin){
//        Schedulers.newThread().createWorker().schedule(()->{
//            System.out.println("--------------------- 第二次开始生产数据 -----------------------");
//            for (int i = 0; i < 10000; i++) {
//                if(i % 100 == 0){
//                    try {
//                        Thread.sleep(28L);
//                    } catch (InterruptedException e) {
//                        e.printStackTrace();
//                    }
//                }
//                TranslatorData request = new TranslatorData();
//
//                String id = UUID.randomUUID() + " --2- "+ i;
//                request.setId(id);
//
//                TestObject object = new TestObject(id,i);
//                request.setMessage(new TestObject[]{object});
//
//                // 作为生产，加入 RingBuffer 处理
//                RingBufferWorkerPoolFactory.getInstance().produceData(request);
//            }
//            System.out.println("--------------------- 第二次结束生产数据 -----------------------");
//        },delayMin, TimeUnit.MINUTES);
    }
}
